package cloud.xiguapi.ubas.analysis.hotpages.model;

import cloud.xiguapi.ubas.model.PageViewCount;
import org.apache.commons.compress.utils.Lists;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.io.Serializable;
import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

/**
 * 热门页面Top N, 实现自定义的处理函数
 *
 * @author 大大大西西瓜皮🍉
 * date: 2021-5-14 下午 02:55
 * desc:
 */
public class TopNHotPages extends KeyedProcessFunction<Long, PageViewCount, TopNHotPagesResult> implements Serializable {

    /**
     * top n的n值, 默认为5
     */
    private Integer topSize = 5;

    /**
     * 定义状态, 保存当前所有PageViewCount到Map中
     */
    private transient MapState<String, Long> pageViewCountMapState;

    public TopNHotPages() {
    }

    public TopNHotPages(Integer topSize) {
        this.topSize = topSize;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        // 初始化状态
        pageViewCountMapState = getRuntimeContext()
                .getMapState(new MapStateDescriptor<>("page-count-map", String.class, Long.class));
    }

    @Override
    public void processElement(PageViewCount value, Context ctx, Collector<TopNHotPagesResult> out) throws Exception {
        // 收集每条数据, 并注册定时器
        // pageViewCountMapState.add(value);
        pageViewCountMapState.put(value.getUrl(), value.getCount());
        ctx.timerService().registerEventTimeTimer(value.getWindowEnd() + 1);

        // 注册一个1min后的定时器, 用于清空状态
        ctx.timerService().registerEventTimeTimer(value.getWindowEnd() + 60 * 1000L);
    }

    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<TopNHotPagesResult> out) throws Exception {
        // 解决同一个URL多次刷榜问题, 需要使用Map保存状态, 而不是List
        // 先判断是否到了窗口关闭清理时间, 如果是直接清空状态返回
        if (timestamp == (ctx.getCurrentKey() + 60 * 1000L)) {
            // 调用清空状态方法
            pageViewCountMapState.clear();
            return;
        }


        // 定时器触发后的处理逻辑
        ArrayList<Map.Entry<String, Long>> pageViewCounts = Lists.newArrayList(pageViewCountMapState.iterator());

        // 排序
        pageViewCounts.sort((o1, o2) -> o2.getValue().compareTo(o1.getValue()));

        List<TopNHotPagesResult.TopNHotPagesDetails> details = new ArrayList<>();

        // 遍历列表, 取top n输出, 需要注意遍历次数, 可能itemViewCounts中数量没有topSize个会造成越界
        for (int i = 0; i < Math.min(topSize, pageViewCounts.size()); i++) {
            // PageViewCount currentPageViewCount = pageViewCounts.get(i);
            Map.Entry<String, Long> currentPageViewCount = pageViewCounts.get(i);
            details.add(TopNHotPagesResult.TopNHotPagesDetails.builder()
                    .url(currentPageViewCount.getKey())
                    .count(currentPageViewCount.getValue())
                    .build());
        }

        TopNHotPagesResult result = TopNHotPagesResult.builder()
                .windowEnd(new Timestamp(timestamp - 1).toString())
                .details(details)
                .build();

        // 输出格式化后的结果
        out.collect(result);
    }
}
